package validator

import (
	"context"
	"github.com/ecodeclub/ekit/slice"
	"golang.org/x/sync/errgroup"
	"gorm.io/gorm"
	"time"
	"webook/pkg/logger"
	"webook/pkg/migrator"
	"webook/pkg/migrator/events"
)

type Validator[T migrator.Entity] struct {
	// 你数据迁移，是不是肯定有 源表 和 目标表；以源表为准的话，base=源表 target=目标表；以目标表为准的话就反过来
	base   *gorm.DB
	target *gorm.DB

	l logger.LoggerV1

	producer events.Producer

	direction string
	batchSize int
	utime     int64
	//<=0 就认为中断 >0就认为睡眠
	sleepInterval time.Duration
	//根据utime和sleepInterval可以实现全量校验和增量校验
	//utime=0 sleepInterval<=0 全量校验 而且数据校验完以后直接退出
	//utime=0 sleepInterval>0 全量校验 而且全量校验完后增量校验
	//utime近期某个时间点 sleepInterval>0 校验近期修改的数据 而且后续继续增量校验
	fromBase      func(ctx context.Context, offset int) (T, error)
	fromBaseBatch func(ctx context.Context, offset, limit int) ([]T, error)
}

func NewValidator[T migrator.Entity](
	base *gorm.DB,
	target *gorm.DB,
	direction string,
	l logger.LoggerV1,
	p events.Producer) *Validator[T] {
	res := &Validator[T]{base: base, target: target,
		l: l, producer: p, direction: direction, batchSize: 100}
	res.fromBase = res.fullFromBase
	return res
}
func (v *Validator[T]) Validate(ctx context.Context) error {
	//并发验证数据
	var eg errgroup.Group
	eg.Go(func() error {
		return v.validateBaseToTarget(ctx)
	})
	eg.Go(func() error {
		return v.validateTargetToBase(ctx)
	})
	return eg.Wait()
}
func (v *Validator[T]) Full() *Validator[T] {
	v.fromBase = v.fullFromBase
	v.fromBaseBatch = v.fullFromBaseBatch
	return v
}
func (v *Validator[T]) Incr() *Validator[T] {
	v.fromBase = v.incrFromBase
	v.fromBaseBatch = v.incrFromBaseBatch
	return v
}

func (v *Validator[T]) Utime(t int64) *Validator[T] {
	v.utime = t
	return v
}

func (v *Validator[T]) SleepInterval(interval time.Duration) *Validator[T] {
	v.sleepInterval = interval
	return v
}
func (v *Validator[T]) fullFromBase(ctx context.Context, offset int) (T, error) {
	dbCtx, cancel := context.WithTimeout(ctx, time.Second)
	defer cancel()
	var src T
	err := v.base.WithContext(dbCtx).Order("id").Offset(offset).First(&src).Error
	return src, err
}
func (v *Validator[T]) incrFromBase(ctx context.Context, offset int) (T, error) {
	dbCtx, cancel := context.WithTimeout(ctx, time.Second)
	defer cancel()
	var src T
	err := v.base.WithContext(dbCtx).Order("utime").Where("utime > ?", v.utime).
		Offset(offset).First(&src).Error
	return src, err
}
func (v *Validator[T]) validateBaseToTarget(ctx context.Context) error {
	offset := 0
	for {
		var src T
		//使用utime进行增量校验
		//err := v.base.WithContext(ctx).Order("utime").Where("utime > ?", v.utime).
		//	Offset(offset).First(&src).Error
		//外部控制哪种模式
		src, err := v.fromBase(ctx, offset)
		//全量校验
		//err := v.base.WithContext(ctx).Order("id").Offset(offset).First(&src).Error
		if err == gorm.ErrRecordNotFound {
			//增量校验时候要考虑一直运行 没有数据的时候 要不要return
			if v.sleepInterval <= 0 {
				return nil
			}
			time.Sleep(v.sleepInterval)
			continue
		}
		if err != nil { //查询出错 考虑offset是否要+1
			v.l.Error("base -> target 查询base失败", logger.Error(err))
			offset++
			continue
		}
		//正常情况
		var dst T
		err = v.target.WithContext(ctx).Where("id = ?", src.ID()).First(&dst).Error
		switch err {
		case gorm.ErrRecordNotFound:
			//target 丢一条消息到kafka
			v.notify(src.ID(), events.InconsistentEventTypeTargetMissing)
		case nil:
			if !src.CompareTo(dst) {
				//丢一条消息到kafka
				v.notify(src.ID(), events.InconsistentEventTypeNEQ)
			}
		default:
			v.l.Error("base -> target 查询base失败", logger.Int64("id", src.ID()), logger.Error(err))
		}
		offset++
	}
	return nil
}
func (v *Validator[T]) validateTargetToBase(ctx context.Context) error {
	offset := 0
	for {
		var ts []T
		err := v.target.WithContext(ctx).Select("id").Order("id").
			Offset(offset).Limit(v.batchSize).Find(&ts).Error
		//不用根据utime去找 因为通过target找base的数据，增量校验的时候，是为了找出来base里没有的数据，
		//这部分数据在base里被删除了，但是target没删，utime没变化，所以不用utime
		//err := v.target.WithContext(ctx).Select("id").Order("id").
		//	Where("utime > ?", v.utime).Offset(offset).
		//	Limit(v.batchSize).Find(&ts).Error
		if err == context.DeadlineExceeded || err == context.Canceled {
			return nil //控制超时
		}
		if err == gorm.ErrRecordNotFound || len(ts) == 0 {
			//增量校验时候要考虑一直运行 没有数据的时候 要不要return
			if v.sleepInterval <= 0 {
				return nil
			}
			time.Sleep(v.sleepInterval)
			continue
		}
		if err != nil {
			v.l.Error("target -> base 查询target失败", logger.Error(err))
			offset += len(ts)
			continue
		}
		//有数据
		var srcTs []T
		ids := slice.Map(ts, func(idx int, t T) int64 {
			return t.ID()
		})
		err = v.base.WithContext(ctx).Select("id").Where("id in ?", ids).
			Find(&srcTs).Error
		if err == gorm.ErrRecordNotFound || len(srcTs) == 0 {
			//没数据
			v.notifyBaseMissing(ts)
			offset += len(ts)
			continue
		}
		if err != nil {
			v.l.Error("target -> base 查询base失败", logger.Error(err))
			offset += len(ts)
			continue
			//能否都认为base里没有数据？也可以但是没有必要
		}
		//找差集
		diff := slice.DiffSetFunc(ts, srcTs, func(src, dst T) bool {
			return src.ID() == dst.ID()
		})
		v.notifyBaseMissing(diff)
		if len(ts) < v.batchSize {
			//增量校验时候要考虑一直运行 没有数据的时候 要不要return
			if v.sleepInterval <= 0 {
				return nil
			}
			time.Sleep(v.sleepInterval)
			continue
		}
		offset += len(ts)
	}
	return nil
}

func (v *Validator[T]) notifyBaseMissing(ts []T) {
	for _, elem := range ts {
		v.notify(elem.ID(), events.InconsistentEventTypeBaseMissing)
	}
}

func (v *Validator[T]) notify(id int64, typ string) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err := v.producer.ProduceInconsistentEvent(ctx, events.InconsistentEvent{
		ID: id, Type: typ, Direction: v.direction,
	})
	if err != nil {
		v.l.Error("发送不一致消息失败", logger.Error(err), logger.String("type", typ), logger.Int64("id", id))
	}
}
